package com.earthea.application.task.quartz;

import cn.hutool.extra.spring.SpringUtil;
import com.earthea.application.task.AbstractTaskService;
import com.earthea.application.task.TaskService;
import com.earthea.application.task.entity.Task;
import com.earthea.application.task.entity.TaskConstants;
import com.earthea.application.utils.BaseLogUtils;
import com.earthea.common.constant.LogConstants;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

public class TaskExecuteQuartzJob extends QuartzJobBean {

    private static final String JOB_NAME = "TaskExecuteQuartzJob";

    @Autowired
    private TaskService taskService;

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        JobDataMap jobDataMap = Optional.of(context).map(JobExecutionContext::getJobDetail).map(JobDetail::getJobDataMap).orElse(new JobDataMap());

        Map<String, String> logParams = Maps.newLinkedHashMap();
        BaseLogUtils.printJobLog(JOB_NAME, LogConstants.BEGIN, logParams);

        String executeBeanId = jobDataMap.getString("executeBeanId");
        logParams.put("executeBeanId", executeBeanId);
        AbstractTaskService service = SpringUtil.getBean(executeBeanId, AbstractTaskService.class);
        if (service == null) {
            BaseLogUtils.printJobLog(JOB_NAME, LogConstants.FAIL, logParams);
            return;
        }

        String threadPoolBeanId = jobDataMap.getString("threadPoolBeanId");
        logParams.put("threadPoolBeanId", threadPoolBeanId);
        ThreadPoolTaskExecutor executor = SpringUtil.getBean(threadPoolBeanId, ThreadPoolTaskExecutor.class);
        if (executor != null) {
            logParams.put("isThreadPool", "true");
        } else {
            logParams.put("isThreadPool", "false");
        }

        int limit = Integer.parseInt(StringUtils.defaultIfBlank(jobDataMap.getString("limit"), String.valueOf(TaskConstants.DEFAULT_ARCHIVING_LIMIT)));
        logParams.put("limit", String.valueOf(limit));

        try {
            int total = 0;
            while (true) {
                List<Task> tasks = taskService.poll(limit);
                if (CollectionUtils.isEmpty(tasks)) {
                    break;
                }
                total += tasks.size();
                List<CompletableFuture<Void>> futures = new ArrayList<>();
                for (Task task : tasks) {
                    if (executor != null) {
                        futures.add(CompletableFuture.runAsync(() -> service.execute(task), executor));
                    } else {
                        service.execute(task);
                    }
                }
                futures.forEach(CompletableFuture::join);
            }
            logParams.put("total", String.valueOf(total));
            BaseLogUtils.printJobLog(JOB_NAME, LogConstants.END, logParams);
        } catch (Exception e) {
            logParams.put("error", e.getMessage());
            BaseLogUtils.printJobLog(JOB_NAME, LogConstants.FAIL, logParams);
        }
    }
}
